package api.beans;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.flink.types.Row;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import utils.DateUtils;
import utils.PropertiesReader;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.Timestamp;
import java.util.Date;
import java.util.UUID;

/**
 * 侧流错误数据的JDBC落地数据库
 * 数据库连接信息由配置文件conf.properties获取
 */
public class ErrorDataJdbcSink2 extends RichSinkFunction<Row> {

    private static Logger logger = LoggerFactory.getLogger(ErrorDataJdbcSink2.class);

    // 声明连接和预编译语句
    Connection connection = null;
    PreparedStatement insertStmt = null;
    private String jobName;

    public ErrorDataJdbcSink2(String jobName) {
        this.jobName = jobName;
    }

    @Override
    public void invoke(Row row, Context context) throws Exception {

        //主键
        String uuid = UUID.randomUUID().toString().replace("-", "");
        insertStmt.setString(1, jobName+ "_" + uuid);
        insertStmt.setString(2, (String) row.getField(0));
        insertStmt.setString(3, (String) row.getField(1));
        insertStmt.setString(4, (String) row.getField(2));
        insertStmt.setString(5, (String) row.getField(3));
        insertStmt.setString(6, (String) row.getField(4));
        insertStmt.setString(7, (String) row.getField(5));
        insertStmt.setString(8, (String) row.getField(6));
        insertStmt.setString(9, (String) row.getField(7));
        insertStmt.setTimestamp(10, new Timestamp((new Date()).getTime()));
        insertStmt.setString(11, DateUtils.formatDate(new Date()));

        insertStmt.execute();
//        insertStmt.addBatch();
//        insertStmt.executeBatch();

    }

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        System.out.println("open......");
        //1.读取配置文件conf.properties获取JDBC连接数据
        String tableName =  PropertiesReader.get("error.data.tablename");
        String driver = PropertiesReader.get("error.data.db.driver");
        String url = PropertiesReader.get("error.data.db.url");
        String username = PropertiesReader.get("error.data.db.username");
        String password = PropertiesReader.get("error.data.db.password");
        System.out.println("Error Data DB JDBC Info: [User]=" + username + ";[Password]=" + password + ";[Driver]=" + driver + ";[Url]:" + url);
        System.out.println("Error Data DB TableName: " + tableName);
        //2.处理insert语句
        StringBuffer sql = new StringBuffer("INSERT INTO ");
        sql.append(tableName);
        sql.append("(f_id,f_realtimetask_id,f_realtimetask_name,f_basictable_id,f_basictable_name,f_kafka_offset,f_source_type,f_dsid,f_error_data,f_storage_time,f_odate) ");
        sql.append("VALUES (?,?,?,?,?,?,?,?,?,?,?)");
        System.out.println("sink sql:" + sql.toString());
        //3.建立JDBC连接
        Class.forName(driver);
        Connection connection = DriverManager.getConnection(url, username, password);
        insertStmt = connection.prepareStatement(sql.toString());
    }

    @Override
    public void close() throws Exception {
        if (insertStmt != null ) {
            insertStmt.close();
        }
        if (connection != null ) {
            connection.close();
        }
        System.out.println("close......");
    }
}
